package com.yonyou.cn.mongodbtest.utils;

import com.mongodb.MongoTimeoutException;
import org.bson.Document;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class SubscriberUtils implements Subscriber {
    //响应数据
    private final List received;
    //错误信息
    private final List<Throwable> errors;
    //等待对象
    private final CountDownLatch latch;
    //订阅器
    private volatile Subscription subscription;
    //是否完成
    private volatile boolean completed;

    public static CopyOnWriteArrayList<Subscription> subscriptions = new CopyOnWriteArrayList<Subscription>();
    AtomicInteger atomicInteger = new AtomicInteger(0);



    public SubscriberUtils(int count) {
        this.received = new ArrayList();
        this.errors = new ArrayList();
        this.latch = new CountDownLatch(count);
    }

    @Override
    public void onSubscribe(final Subscription s) {
        subscription = s;
        subscriptions.add(subscription);
//        subscription.request(Integer.MAX_VALUE);
    }
    @Override
    public void onNext(Object o) {
        received.add(o);
    }

    @Override
    public void onError(final Throwable t) {
        errors.add(t);
        onComplete();
    }

    @Override
    public void onComplete() {
        System.out.println("haha");
        atomicInteger.getAndIncrement();
        completed = true;
        latch.countDown();
    }

    public Subscription getSubscription() {
        return subscription;
    }

    public List getReceived() {
        return received;
    }

    public Throwable getError() {
        if (errors.size() > 0) {
            return errors.get(0);
        }
        return null;
    }

    public boolean isCompleted() {
        return completed;
    }

    /**
     * 阻塞一定时间等待结果
     *
     * @param timeout
     * @param unit
     * @return
     * @throws Throwable
     */
    public List get(final long timeout, final TimeUnit unit) throws Throwable {
        return await(timeout, unit).getReceived();
    }

    /**
     * 一直阻塞等待请求完成
     *
     * @return
     * @throws Throwable
     */
    public SubscriberUtils await() throws Throwable {
        return await(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    }

    /**
     * 阻塞一定时间等待完成
     *
     * @param timeout
     * @param unit
     * @return
     * @throws Throwable
     */
    public SubscriberUtils await(final long timeout, final TimeUnit unit) throws Throwable {
//        System.out.println(subscriptions.size());
        subscriptions.forEach(subscription->{
            subscription.request(Integer.MAX_VALUE);
        });
//        System.out.println("等会儿");
//        long a1 = System.currentTimeMillis();
        /*boolean flag = true;
        while(flag){
            if(atomicInteger.get()==5){
                flag = false;
            }
        }*/
        if (!latch.await(timeout, unit)) {
            throw new MongoTimeoutException("Publisher onComplete timed out");
        }

//        long a2 = System.currentTimeMillis();
//        System.out.println("等了======="+(a2-a1)/1000+"s");
        if (!errors.isEmpty()) {
            throw errors.get(0);
        }
        return this;
    }
}
